package com.azzhu.bigdata;

import com.azzhu.bigdata.bean.Stu;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.StringUtils;

/**
 * @author azzhu
 * @create 2022-04-09 23:28:12
 * @description
 */
public class MapStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop105:8020/flink/ck/demo1");

        env.socketTextStream("hadoop105", 9999)
                .map(new MapFunction<String, Stu>() {
                    @Override
                    public Stu map(String msg) throws Exception {
                        String[] split = msg.split(",");
                        return new Stu(split[0],split[1],Long.parseLong(split[2]));
                    }
                })
                .keyBy(new KeySelector<Stu, String>() {
                    @Override
                    public String getKey(Stu stu) throws Exception {
                        return stu.getId();
                    }
                })
                .process(new KeyedProcessFunction<String, Stu, Stu>() {
                    MapState<String,Stu> stuMapState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        MapStateDescriptor<String,Stu> stringStuMapStateDescriptor = new MapStateDescriptor<String, Stu>("stuMapState",String.class,Stu.class);
                        stuMapState = getRuntimeContext().getMapState(stringStuMapStateDescriptor);
                    }

                    @Override
                    public void processElement(Stu stu, KeyedProcessFunction<String, Stu, Stu>.Context context, Collector<Stu> collector) throws Exception {
                        String id = stu.getId();
                        if(stuMapState == null || stuMapState.get(id) == null) {
                            System.out.println(stu + "第一次进入flink");
                            stuMapState.put(id,stu);
                            collector.collect(stu);
                            return;
                        }

                        //非第一次进入,修改年龄
                        Stu stu2 = stuMapState.get(id);
                        if(stu2 != null && id.equals(stu2.getId())) {
                            System.out.println("学号相同,修改年龄值,新信息:"+stu);
                            stuMapState.put(id,stu);
                            collector.collect(stu);
                        }
                    }
                })
                .print();

        env.execute();
    }
}
